package com.xmmxjy.hbase.reposity;

import com.xmmxjy.common.util.Tools;
import com.xmmxjy.hbase.entity.*;
import com.xmmxjy.hbase.util.BeanUtil;
import com.xmmxjy.hbase.util.HBasePageModel;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Repository;

import javax.tools.Tool;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.xmmxjy.hbase.reposity.UserRepository.CF_INFO;
import static jdk.nashorn.internal.runtime.regexp.joni.Config.log;

/**
 * Created by xmm on 2017/2/16.
 */
@Repository
public class DwUserReposity {

    private static Logger log = LoggerFactory.getLogger(DwUserReposity.class);

    public static String TYPE_ADD = "insert";

    public static String TYPE_UPDATE = "update";

    public static String TYPE_DELETE = "delete";

    @Autowired
    private HbaseTemplate hbaseTemplate;

    @Autowired
    private Configuration hbaseConfiguration;

    //用户分类的表名
    private String categeroyName = "user_feature_category";
    //用户主题词的表名
    private String topicName = "user_feature_topic";
    //用户频道的表名
    private String channelName = "user_preference_content";
    //列簇名
    private String familyName = "uf";
    public static byte[] UF_BYTE = Bytes.toBytes("uf");

    private byte[] qUser = Bytes.toBytes("user");
    private byte[] qEmail = Bytes.toBytes("email");
    private byte[] qPassword = Bytes.toBytes("password");


    private byte[] username = Bytes.toBytes("username");
    private byte[] email = Bytes.toBytes("email");
    private byte[] clientId = Bytes.toBytes("clientId");
    private byte[] phone = Bytes.toBytes("phone");
    //性别
    private byte[] sex = Bytes.toBytes("sex");

    //归属地,根据地域推荐
    private byte[] home = Bytes.toBytes("home");
    //出生年月日，根据年龄推荐
    private byte[] birthday = Bytes.toBytes("birthday");
    //年龄
    private byte[] age = Bytes.toBytes("age");
    //星座，根据星座推荐
    private byte[] constellation = Bytes.toBytes("constellation");
    //所在地，根据地域推荐
    private byte[] location= Bytes.toBytes("location");
    //设备标识号
    private byte[] imei = Bytes.toBytes("imei");
    private byte[] os = Bytes.toBytes("os");

    private byte[] model = Bytes.toBytes("model");

    private byte[] operator = Bytes.toBytes("operator");
    private byte[] qualifications = Bytes.toBytes("qualifications");
    private byte[] isMarry = Bytes.toBytes("isMarry");
    private byte[] profession = Bytes.toBytes("profession");
    private byte[] interest = Bytes.toBytes("interest");
    private byte[] keyWords = Bytes.toBytes("keyWords");
    private byte[] questionList = Bytes.toBytes("questionList");
    private byte[] typeList = Bytes.toBytes("typeList");

    private String baseUserName = "user_base_info";

    /**
     * 根据手机或者客户端号查询喜欢的分类列表
     * @param phone
     * @return
     */
    public List<DwCategory> findCateGoryList(String phone){

        DwUser du = hbaseTemplate.get(categeroyName, phone, new RowMapper<DwUser>() {

            @Override
            public DwUser mapRow(Result result, int i) throws Exception {
                List<DwCategory> categoryList = new ArrayList<DwCategory>();
                for (KeyValue rowKV : result.raw()) {
                    DwCategory dc = new DwCategory();
                    dc.setCategoryCode(Bytes.toString(rowKV.getQualifier()));
                    dc.setCategoryName(BeanUtil.getCategoryName(dc.getCategoryCode()));
                    dc.setCategroyWeight(Double.parseDouble(Bytes.toString(rowKV.getValue())));
                    categoryList.add(dc);
                }
                DwUser du = new DwUser();
                du.setCategoryList(categoryList);
                return du;
            }
        });
        if (du != null) {
            return BeanUtil.getCategoryList(du.getCategoryList(),BeanUtil.getCategoryMaxWeight(du.getCategoryList()),10);
        }
        return null;
    }

    /**
     * 根据手机号或者客户端号查询主题词列表
     * @param phone
     * @return
     */
    public List<DwTopic> findTopicList(String phone) {
        DwUser du = hbaseTemplate.get(topicName, phone, new RowMapper<DwUser>() {

            @Override
            public DwUser mapRow(Result result, int i) throws Exception {
                List<DwTopic> topicList = new ArrayList<DwTopic>();
                for (KeyValue rowKV : result.raw()) {
                    DwTopic dt = new DwTopic();
                    dt.setTopicName(Bytes.toString(rowKV.getQualifier()));
                    dt.setTopicWeight(Double.parseDouble(Bytes.toString(rowKV.getValue())));
                    topicList.add(dt);
                }
                DwUser du = new DwUser();
                du.setTopicList(topicList);
                return du;
            }
        });
        if (du != null) {
            return BeanUtil.getTopicList(du.getTopicList(),BeanUtil.getMaxWeight(du.getTopicList()),10);
        }
        return null;
    }

    /**
     * 根据手机号或者客户端号查询频道列表
     * @param phone
     * @return
     */
    public List<DwChannel> findDwChannelList(String phone) {
        DwUser du = hbaseTemplate.get(channelName, phone, new RowMapper<DwUser>() {
            @Override
            public DwUser mapRow(Result result, int i) throws Exception {
                List<DwChannel> channelList = new ArrayList<DwChannel>();
                for (KeyValue rowKV : result.raw()) {
                    DwChannel dc = new DwChannel();
                    dc.setChannelName(Bytes.toString(rowKV.getQualifier()));
                    dc.setChannelWeight(Double.parseDouble(Bytes.toString(rowKV.getValue())));
                    channelList.add(dc);
                }
                DwUser user = new DwUser();
                user.setChannelList(channelList);
                return user;
            }
        });
        if (du != null) {
            return BeanUtil.getChannelList(du.getChannelList(),BeanUtil.getCategoryMaxWeight(du.getCategoryList()),10);
        }
        return null;
    }

    /**
     * 根据分类编码查询用户列表
     * @param categeroyCode
     * @return
     */
    public List<DwUser> findUserByCategory(String categeroyCode) {
        Scan scan = new Scan();
        scan.addColumn(UF_BYTE,Bytes.toBytes(categeroyCode));
        return hbaseTemplate.find(categeroyName, scan, new RowMapper<DwUser>() {
            @Override
            public DwUser mapRow(Result result, int i) throws Exception {
                DwUser dw = new DwUser();
                dw.setPhone(Bytes.toString(result.getRow()));
                return dw;
            }
        });
    }

    /**
     * 根据主题词查询用户列表
     * @param topicName
     * @return
     */
    public List<DwUser> findUserByTopic(String topicName) {
        Scan scan = new Scan();
        scan.addColumn(UF_BYTE,Bytes.toBytes(topicName));
        return hbaseTemplate.find(topicName, scan, new RowMapper<DwUser>() {
            @Override
            public DwUser mapRow(Result result, int i) throws Exception {
                DwUser dw = new DwUser();
                dw.setPhone(Bytes.toString(result.getRow()));
                return dw;
            };
        });
    }

    /**
     * 根据栏目名称或者编码查询用户列表
     * @param channelName
     * @return
     */
    public List<DwUser> findUserByChannel(String channelName) {
        Scan scan = new Scan();
        scan.addColumn(UF_BYTE,Bytes.toBytes(channelName));
        return hbaseTemplate.find(channelName, scan, new RowMapper<DwUser>() {
            @Override
            public DwUser mapRow(Result result, int i) throws Exception {
                DwUser dw = new DwUser();
                dw.setPhone(Bytes.toString(result.getRow()));
                return dw;
            }
        });
    }

    /**
     * 保存用户基本信息
     * @param base
     * @return
     */
    public DwBaseUser saveBaseUser(final DwBaseUser base) {
        return hbaseTemplate.execute(baseUserName, new TableCallback<DwBaseUser>() {
            public DwBaseUser doInTable(HTableInterface table) throws Throwable {
                Put p = new Put(Bytes.toBytes(base.getId()));
                if (Tools.notEmpty(base.getClientId())) {
                    p.add(UF_BYTE,clientId,Bytes.toBytes(base.getClientId()));
                }
                if (Tools.notEmpty(base.getPhone())) {
                    p.add(UF_BYTE,phone,Bytes.toBytes(base.getPhone()));
                }
                if (Tools.notEmpty(base.getEmail())) {
                    p.add(UF_BYTE,email,Bytes.toBytes(base.getEmail()));
                }
                if (Tools.notEmpty(base.getUsername())) {
                    p.add(UF_BYTE, username, Bytes.toBytes(base.getUsername()));
                }
                if (Tools.notEmpty(base.getSex())) {
                    p.add(UF_BYTE, sex, Bytes.toBytes(base.getSex()));
                }
                if (Tools.notEmpty(base.getHome())) {
                    p.add(UF_BYTE, home, Bytes.toBytes(base.getHome()));
                }
                if (Tools.notEmpty(base.getBirthday())) {
                    p.add(UF_BYTE, birthday, Bytes.toBytes(base.getBirthday()));
                }
                if (Tools.notEmpty(base.getAge())) {
                    p.add(UF_BYTE, age, Bytes.toBytes(base.getAge()));
                }
                if (Tools.notEmpty(base.getConstellation())) {
                    p.add(UF_BYTE, constellation, Bytes.toBytes(base.getConstellation()));
                }
                if (Tools.notEmpty(base.getLocation())) {
                    p.add(UF_BYTE, location, Bytes.toBytes(base.getLocation()));
                }
                if (Tools.notEmpty(base.getImei())) {
                    p.add(UF_BYTE, imei, Bytes.toBytes(base.getImei()));
                }
                if (Tools.notEmpty(base.getOs())) {
                    p.add(UF_BYTE, os, Bytes.toBytes(base.getOs()));
                }
                if (Tools.notEmpty(base.getModel())) {
                    p.add(UF_BYTE, model, Bytes.toBytes(base.getModel()));
                }
                if (Tools.notEmpty(base.getOperator())) {
                    p.add(UF_BYTE, operator, Bytes.toBytes(base.getOperator()));
                }
                if (Tools.notEmpty(base.getQualifications())) {
                    p.add(UF_BYTE, qualifications, Bytes.toBytes(base.getQualifications()));
                }
                if (Tools.notEmpty(base.getIsMarry())) {
                    p.add(UF_BYTE, isMarry, Bytes.toBytes(base.getIsMarry()));
                }
                if (Tools.notEmpty(base.getProfession())) {
                    p.add(UF_BYTE, profession, Bytes.toBytes(base.getProfession()));
                }
                if (Tools.notEmpty(base.getInterest())) {
                    p.add(UF_BYTE, interest, Bytes.toBytes(base.getInterest()));
                }
                if (Tools.notEmpty(base.getKeyWords())) {
                    p.add(UF_BYTE, keyWords, Bytes.toBytes(base.getKeyWords()));
                }
                if (Tools.notEmpty(base.getQuestionList())) {
                    p.add(UF_BYTE, questionList, Bytes.toBytes(base.getQuestionList()));
                }
                if (Tools.notEmpty(base.getTypeList())) {
                    p.add(UF_BYTE, typeList, Bytes.toBytes(base.getTypeList()));
                }
                if (Tools.notEmpty(base.getSumPoint())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumPoint"), Bytes.toBytes(base.getSumPoint()));

                }
                if (Tools.notEmpty(base.getConsumePonit())) {
                    p.add(UF_BYTE, Bytes.toBytes("consumePonit"), Bytes.toBytes(base.getConsumePonit()));

                }
                if (Tools.notEmpty(base.getSumCollection())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumCollection"), Bytes.toBytes(base.getSumCollection()));

                }
                if (Tools.notEmpty(base.getSumAsk())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumAsk"), Bytes.toBytes(base.getSumAsk()));

                }
                if (Tools.notEmpty(base.getSumAnswer())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumAnswer"), Bytes.toBytes(base.getSumAnswer()));

                }
                if (Tools.notEmpty(base.getSumComment())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumComment"), Bytes.toBytes(base.getSumComment()));

                }
                if (Tools.notEmpty(base.getSumPoint())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumLike"), Bytes.toBytes(base.getSumPoint()));

                }
                if (Tools.notEmpty(base.getSumRead())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumRead"), Bytes.toBytes(base.getSumRead()));

                }
                if (Tools.notEmpty(base.getSumShare())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumShare"), Bytes.toBytes(base.getSumShare()));

                }
                if (Tools.notEmpty(base.getSumUnInterested())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumUnInterested"), Bytes.toBytes(base.getSumUnInterested()));

                }
                if (Tools.notEmpty(base.getSumUnLike())) {
                    p.add(UF_BYTE, Bytes.toBytes("sumUnLike"), Bytes.toBytes(base.getSumUnLike()));

                }
                table.put(p);
                return base;
            }
        });
    }
    //更新用户信息
    public DwBaseUser deleteBaseUser(final DwBaseUser base) {
        return hbaseTemplate.execute(baseUserName, new TableCallback<DwBaseUser>() {
            @Override
            public DwBaseUser doInTable(HTableInterface table) throws Throwable {
                Delete d = new Delete(Bytes.toBytes(base.getId()));
                table.delete(d);
                return base;
            }
        });
    }

    /**
     * 根据用户ID查询用户
     * @param id
     * @return
     */
    public DwBaseUser getBaseUser(String id) {
        return hbaseTemplate.get(baseUserName, id, new RowMapper<DwBaseUser>() {
            @Override
            public DwBaseUser mapRow(Result result, int i) throws Exception {
                DwBaseUser base = new DwBaseUser();
                base.setId(Bytes.toString(result.getRow()));
                base.setClientId(Bytes.toString(result.getValue(UF_BYTE,clientId)));
                base.setPhone(Bytes.toString(result.getValue(UF_BYTE,phone)));
                base.setBirthday(Bytes.toString(result.getValue(UF_BYTE,birthday)));
                base.setSex(Bytes.toString(result.getValue(UF_BYTE,sex)));
                base.setEmail(Bytes.toString(result.getValue(UF_BYTE,email)));
                base.setHome(Bytes.toString(result.getValue(UF_BYTE,home)));
                base.setAge(Bytes.toString(result.getValue(UF_BYTE,age)));
                base.setUsername(Bytes.toString(result.getValue(UF_BYTE,username)));
                base.setConstellation(Bytes.toString(result.getValue(UF_BYTE,constellation)));
                base.setLocation(Bytes.toString(result.getValue(UF_BYTE,location)));
                base.setImei(Bytes.toString(result.getValue(UF_BYTE,imei)));
                base.setOs(Bytes.toString(result.getValue(UF_BYTE,os)));
                base.setOperator(Bytes.toString(result.getValue(UF_BYTE,operator)));
                base.setModel(Bytes.toString(result.getValue(UF_BYTE,model)));
                base.setQualifications(Bytes.toString(result.getValue(UF_BYTE,qualifications)));
                base.setProfession(Bytes.toString(result.getValue(UF_BYTE,profession)));
                base.setInterest(Bytes.toString(result.getValue(UF_BYTE,interest)));
                base.setKeyWords(Bytes.toString(result.getValue(UF_BYTE,keyWords)));
                base.setQuestionList(Bytes.toString(result.getValue(UF_BYTE,questionList)));
                base.setTypeList(Bytes.toString(result.getValue(UF_BYTE,typeList)));
                return base;
            }
        });
    }

    public List<DwBaseUser> findBaseUser(String un,int pageSize) {
        final Scan scan = new Scan();
        FilterList filterList = new FilterList();
        Filter pageFilter = new PageFilter(pageSize);
        filterList.addFilter(pageFilter);
        if (Tools.notEmpty(un)) {
            SingleColumnValueFilter filter = new SingleColumnValueFilter(UF_BYTE, username, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(un));
            filterList.addFilter(filter);
        }
        scan.setFilter(filterList);


        return hbaseTemplate.find(baseUserName,scan,new RowMapper<DwBaseUser>() {
            @Override
            public DwBaseUser mapRow(Result result, int i) throws Exception {
                System.out.println("i : " + i);
                DwBaseUser base = new DwBaseUser();
                base.setId(Bytes.toString(result.getRow()));
                base.setClientId(Bytes.toString(result.getValue(UF_BYTE,clientId)));
                base.setUsername(Bytes.toString(result.getValue(UF_BYTE,username)));
                return base;
            }
        });
    }
    public HBasePageModel<DwBaseUser> findBaseUser(String un,byte[] startRowKey, byte[] endRowKey,HBasePageModel<DwBaseUser> pageModel) {
        if(pageModel == null) {
            pageModel = new HBasePageModel<DwBaseUser>(10);
        }
        FilterList filterList = new FilterList();
        pageModel.initStartTime();
        pageModel.initEndTime();
        int tempPageSize = pageModel.getPageSize();
        boolean isEmptyStartRowKey = false;

        if(startRowKey != null) {
            isEmptyStartRowKey = true;
            pageModel.setPageStartRowKey(startRowKey);
            tempPageSize += 1;
        } else {
            if(pageModel.getPageEndRowKey() != null) {
                pageModel.setPageStartRowKey(pageModel.getPageEndRowKey());
            }
            //从第二页开始，每次都多取一条记录，因为第一条记录是要删除的。
        }
        Scan scan = new Scan();
        if (pageModel.getPageStartRowKey() != null) {
            scan.setStartRow(pageModel.getPageStartRowKey());
        }
        if(endRowKey != null) {
            scan.setStopRow(endRowKey);
        }
        PageFilter pageFilter = new PageFilter(tempPageSize);
        filterList.addFilter(pageFilter);
        scan.setFilter(filterList);
        List<DwBaseUser> userList = hbaseTemplate.find(baseUserName,scan,new RowMapper<DwBaseUser>() {
            @Override
            public DwBaseUser mapRow(Result result, int i) throws Exception {
                System.out.println("i : " + i);
                DwBaseUser base = new DwBaseUser();
                base.setId(Bytes.toString(result.getRow()));
                base.setClientId(Bytes.toString(result.getValue(UF_BYTE,clientId)));
                base.setUsername(Bytes.toString(result.getValue(UF_BYTE,username)));
                return base;
            }
        });

        int pageIndex = pageModel.getPageIndex() + 1;
        pageModel.setPageIndex(pageIndex);
        if(userList.size() > 0) {
            //获取本次分页数据首行和末行的行键信息
            byte[] pageStartRowKey = Bytes.toBytes(userList.get(0).getId());
            byte[] pageEndRowKey = Bytes.toBytes(userList.get(userList.size() - 1).getId());
            pageModel.setPageStartRowKey(pageStartRowKey);
            pageModel.setPageEndRowKey(pageEndRowKey);
            if (userList.size() == pageModel.getPageSize() + 1){
                pageModel.setResultList(userList.subList(1,userList.size()));
            } else {
                pageModel.setResultList(userList);
            }
        }
        return pageModel;
    }
  /*  public int getPages(String tableName, String ddate, int pageSize) {

        enableAggregation(tableName);
        int total = 0;
        try {
            HTable hTable = (HTable) getHTable(tableName);
            Scan scan = new Scan();
            Filter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL,
                    new SubstringComparator(ddate));
            scan.setFilter(rowFilter);
            AggregationClient aggregation = new AggregationClient(hbaseConfiguration);
            Long count = aggregation.rowCount(hTable,
                    new LongColumnInterpreter(), scan);
            total = count.intValue();
            hTable.close();
        } catch (Throwable e) {
            // TODO Auto-generated catch block
            log.error(e.getMessage(),e);
        }
        return (total % pageSize == 0) ? total / pageSize
                : (total / pageSize) + 1;
    }
    @SuppressWarnings("resource")
    private void enableAggregation(String tableName) {
        String coprocessorName = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";
        try {
            HBaseAdmin admin = new HBaseAdmin(hbaseConfiguration);
            HTableDescriptor htd = admin.getTableDescriptor(Bytes
                    .toBytes(tableName));
            List<String> coprocessors = htd.getCoprocessors();
            if (coprocessors != null && coprocessors.size() > 0) {
                return;
            } else {
                admin.disableTable(tableName);
                htd.addCoprocessor(coprocessorName);
                admin.modifyTable(tableName, htd);
                admin.enableTable(tableName);
            }
        } catch (TableNotFoundException e) {
            // TODO Auto-generated catch block
            log.error(e.getMessage());
        } catch (MasterNotRunningException e) {
            // TODO Auto-generated catch block
            log.error(e.getMessage());
        } catch (ZooKeeperConnectionException e) {
            // TODO Auto-generated catch block
            log.error(e.getMessage());
        } catch (IOException e) {
            // TODO Auto-generated catch block
            log.error(e.getMessage());
        }
    }*/

    /**
     * 根据性别数量
     * @param sexValue
     * @return
     */
    public int getCountbySex(String sexValue) {
        Scan scan = new Scan();
        FilterList filterList = new FilterList();
        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(UF_BYTE, sex, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(sexValue));
        filter.setFilterIfMissing(true);
        filterList.addFilter(filter);
        scan.setFilter(filterList);
        List <DwBaseUser> baseUsers = hbaseTemplate.find(baseUserName, scan, new RowMapper<DwBaseUser>() {

            @Override
            public DwBaseUser mapRow(Result result, int i) throws Exception {
                DwBaseUser user = new DwBaseUser();
                user.setId(Bytes.toString(result.getRow()));
                return user;
            }
        });
        if (baseUsers != null) {
            return baseUsers.size();
        }
        return 0;
    }

    /**
     * 查询用户总数
     * @return
     */
    public int getCountUser() {

        List <DwBaseUser> baseUsers = hbaseTemplate.find(baseUserName, "uf", new RowMapper<DwBaseUser>() {

            @Override
            public DwBaseUser mapRow(Result result, int i) throws Exception {
                DwBaseUser user = new DwBaseUser();
                user.setId(Bytes.toString(result.getRow()));
                return user;
            }
        });
        if (baseUsers != null) {
            return baseUsers.size();
        }
        return 0;
    }

    /**
     * 根据年龄得到数量
     * @param min
     * @param max
     * @return
     */
    public int getCountAge(String min,String max) {

        Scan scan = new Scan();
        FilterList filterList = new FilterList();
        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(UF_BYTE, age, CompareFilter.CompareOp.GREATER_OR_EQUAL, Bytes.toBytes(min));
        SingleColumnValueExcludeFilter filter2 = new SingleColumnValueExcludeFilter(UF_BYTE, age, CompareFilter.CompareOp.LESS, Bytes.toBytes(max));
        filter.setFilterIfMissing(true);
        filterList.addFilter(filter);
        filterList.addFilter(filter2);
        scan.setFilter(filterList);
        List <DwBaseUser> baseUsers = hbaseTemplate.find(baseUserName, scan, new RowMapper<DwBaseUser>() {

            @Override
            public DwBaseUser mapRow(Result result, int i) throws Exception {
                DwBaseUser user = new DwBaseUser();
                user.setId(Bytes.toString(result.getRow()));
                return user;
            }
        });
        if (baseUsers != null) {
            return baseUsers.size();
        }
        return 0;
    }
    /**
     * 求各地域所在人数
     */
    public Map<String,String> areaList() {
        return hbaseTemplate.get("user_area_sum","areaSum",new RowMapper<Map<String,String>>(){
            @Override
            public Map<String,String> mapRow(Result result, int i) throws Exception {
                Map<String,String> map = new HashMap<String, String>();
                for(KeyValue kv :result.raw()){
                    String qu = new String(kv.getQualifier());
                    String count = new String(kv.getValue());
                    map.put(qu,count);
                }
                return map;
            }
        });
    }
}
